package com.java.springboot.elasticsearch.service.impl;

import com.alibaba.fastjson.JSON;
import com.java.springboot.elasticsearch.common.RedisKey;
import com.java.springboot.elasticsearch.mapper.ReadBookPdMapper;
import com.java.springboot.elasticsearch.model.ReadBookPd;
import com.java.springboot.elasticsearch.model.ReadBookPdExample;
import com.java.springboot.elasticsearch.service.ReadBookPdService;
import com.java.springboot.elasticsearch.service.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author zhangtengfei
 * @date 2021/9/30 11:30
 */
@Slf4j
@Service
public class ReadBookPdServiceImpl implements ReadBookPdService {

    @Resource
    private ReadBookPdMapper readBookPdMapper;

    @Resource
    private RedisService redisService;

    @Resource
    private RestHighLevelClient restHighLevelClient;

    ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(10, 100, 100L, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(1000));

    @Override
    public Integer indexAll(String index) {
        ReadBookPdExample example = new ReadBookPdExample();
        ReadBookPdExample.Criteria criteria = example.createCriteria();
        // 正常上架的书
        criteria.andStatusEqualTo(1);
        int count = readBookPdMapper.countByExample(example);
        // 每次写入100条数据到es
        int pageSize = 100;
        int pages = (count / pageSize);
        // 最多只让搜前100页的（京东商品页也是只有100页）
        pages = Math.min(pages, 100);
        // 100条数据 - 1页; 101条数据 - 2页
        if (count % pageSize != 0) {
            pages++;
        }
        log.info("数据库中记录总数: {}, pages: {}", count, pages);
        for (int pageNo = 1; pageNo <= pages; pageNo++) {
            final int tmpPageNo = pageNo;
            threadPoolExecutor.execute(() -> {
                // 第一页开始：0  第二页开始：100 ...
                int offset = (tmpPageNo - 1) * pageSize;
                List<ReadBookPd> readBookPdList = readBookPdMapper.getReadBookPdList(offset, pageSize);
                if (!CollectionUtils.isEmpty(readBookPdList)) {
                    // 创建es索引
                    addToEs(readBookPdList, index);
                }
            });
        }
        // 线程池不能关闭，重新构建索引时还会接着使用
//        try {
//            threadPoolExecutor.shutdown();
//            while (!threadPoolExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
//                log.info("等待索引重建完成.....");
//            }
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
        return count;
    }

    @Override
    public Boolean reIndexAll(String index) {
        log.info("开始对books索引进行全量重建");
        String updateIndex = redisService.get(RedisKey.cacheKeys("update"), String.class);
        if (StringUtils.isEmpty(updateIndex)) {
            updateIndex = index + "_update";
        }
        String currentIndex = redisService.get(RedisKey.cacheKeys("current"), String.class);
        if (StringUtils.isEmpty(currentIndex)) {
            currentIndex = index;
        }
        log.info("当前备份的索引集合为{}，正在服务中的索引集合为{}", updateIndex, currentIndex);
        // 更新备份索引，更新完进行切换。
        indexAll(updateIndex);
        log.info("对books索引全量重建完成,进行集合的切换");
        redisService.set(RedisKey.cacheKeys("update"), currentIndex);
        redisService.set(RedisKey.cacheKeys("current"), updateIndex);
        log.info("切换成功,下次再备份使用索引: {}, 正在服务中的索引: {}", currentIndex, updateIndex);
        return true;
    }

    /**
     * ------------------------------------------------- 分割线 -------------------------------------------------
     * 将所有字段加入es中
     *
     * @param readBookPdList
     */
    private void addToEs(List<ReadBookPd> readBookPdList, String index) {
        BulkRequest request = new BulkRequest();
        readBookPdList.forEach(readBookPd -> {
            String jsonStr = JSON.toJSONString(readBookPd);
            request.add(new IndexRequest(index).id(readBookPd.getId().toString()).source(jsonStr, XContentType.JSON));
        });
        /**
         * 同步执行
         */
        try {
            restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error("bulk error {}", e.getMessage(), e);
        }
    }

    /**
     * 这里可以指定字段添加到es中
     *
     * @param readBookPdList
     */
    private void addToEsWithMap(List<ReadBookPd> readBookPdList, String index) {
        BulkRequest request = new BulkRequest();
        readBookPdList.forEach(readBookPd -> {
            Map<String, Object> map = new HashMap<>();
            map.put("id", readBookPd.getId());
            map.put("name", readBookPd.getName());
            map.put("enName", readBookPd.getEnName());
            map.put("author", readBookPd.getAuthor());
            map.put("imgurl", readBookPd.getImgurl());
            map.put("createTime", readBookPd.getCreateTime().getTime());
            map.put("status", readBookPd.getStatus());
            map.put("description", readBookPd.getDescription());
            map.put("price", readBookPd.getPrice());
            map.put("category", readBookPd.getCategory());
            map.put("commentNum", readBookPd.getCommentNum());
            // 如果自己定义了score字段，这里就会有一个计算得分的模块
            // map.put("score","1231");
            request.add(new IndexRequest(index).id(readBookPd.getId().toString()).source(map));
        });
        /**
         * 同步执行
         */
        try {
            restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error("bulk error {}", e.getMessage(), e);
        }
    }
}
